package com.enjoyor.soa.traffic.frame.support.amq;

import com.enjoyor.soa.traffic.util.helper.LoggerHelper;

/**
 * 冗余数据接受器，在后台数据处理程序启动时调用，用于清空可能存在的 AMQ 消息队列累积数据 (避免队列中已有数据过期对后续处理的影响)，
 * 使用代码示例如下：
 * 
 * <pre>
 * RedundantMessageReceiver ram = new RedundantMessageReceiver(url, queue);
 * ram.consume();
 * </pre>
 * 
 * @author WY
 * @reviser WY
 * @date 2015-4-8
 * @see AMQReceiver
 * @see AMQStatistics
 */
public class RedundantMessageReceiver {
	private AMQReceiver receiver;
	private AMQStatistics statistics;
	private String url;
	private String queue;

	/**
	 * 冗余数据接受器构造器，负责配置AMQ信息服务与消息接收器
	 * 
	 * @param url
	 *            指定的队列地址
	 * @param queue
	 *            指定的队列名称
	 */
	public RedundantMessageReceiver(String url, String queue) {
		this.url = url;
		this.queue = queue;
		this.statistics = new AMQStatistics(url, queue);
		this.receiver = new AMQReceiver(url, queue);
	}

	private void initialize() {
		this.statistics.initialize();
		this.receiver.initialize();
	}

	private void end() {
		this.statistics.endEngine();
		this.receiver.endEngine();
	}

	private Object receive(int timeout) {
		return receiver.receiveMessage(timeout);
	}

	private Object getStatistics(String name) {
		return this.statistics.getStatistics(name);
	}

	/**
	 * 取出指定队列的所有数据，并记录日志
	 */
	public void consume() {
		long t1 = System.currentTimeMillis();
		initialize();
		Object st = getStatistics("size");
		long size = st == null ? 0 : (Long) st;
		long count = 0;
		while (count < size) {
			Object message = receive(100);
			if (message != null) {
				count++;
			}
		}
		long t2 = System.currentTimeMillis();
		LoggerHelper.LOG.info(url + "/queue:" + queue + " 队列剩余"
				+ (size > 0 ? --size : 0) + "条数据已取出" + "(time=" + (t2 - t1)
				+ "ms)");
		end();
	}
}
